﻿//========================================================= 
/**@file namedpipe_async_svr.h
 * @brief 通过管道收发监控参数,以信号的方式通知客户端获取数据的监控项
 * 
 * @date 2018-10-11   11:09:25
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_NAMEDPIPE_ASYNC_SVR_H_
#define _LIBZHOUYB_NAMEDPIPE_ASYNC_SVR_H_
//--------------------------------------------------------- 
#include "namedpipe_svr.h"

#include "../lmdb/lmdb_event.h"
using zhou_yb::wrapper::lmdb_extractor::lmdb_env;
using zhou_yb::wrapper::lmdb_extractor::lmdb_string_converter;
using zhou_yb::wrapper::lmdb_extractor::lmdb_helper;
using zhou_yb::wrapper::lmdb_extractor::semaphore_signal;
using zhou_yb::wrapper::lmdb_extractor::lmdb_event_agent;
//--------------------------------------------------------- 
namespace zhou_yb {
namespace wrapper {
namespace zabbix_extractor {
//--------------------------------------------------------- 
/// 异步的管道名称
#ifndef ZABBIX_ASYNC_NAMEDPIPE_NAME
#   define ZABBIX_ASYNC_NAMEDPIPE_NAME "ZabbixAsync"
#endif
//--------------------------------------------------------- 
/**
 * @brief 异步的管道监控项服务端
 *
 * @description 通过管道接收客户端的监控请求,将每个监控项的监控数据写入lmdb,使用信号通知客户端获取数据
 *
 * @date 2018-10-12 15:52:08
 */
class namedpipe_async_svr : public converter_dispatcher
{
public:
    typedef converter_dispatcher::callback_type callback_type;

    /**
     * @brief 返回对应操作指令的事件KEY
     * @date 2018-10-12 16:20:04
     */
    static string get_key(const string& cmd, const string& param)
    {
        MD5_Provider md5Provider;
        ByteBuilder md5;
        ByteBuilder md5Ascii;
        md5Provider.Update(ByteArray(cmd.c_str(), cmd.length()));
        md5Provider.Update(ByteArray(param.c_str(), param.length()));
        md5Provider.Final(md5);
        ByteConvert::ToAscii(md5, md5Ascii);
        string key(md5Ascii.GetString(), md5Ascii.GetLength());
        return key;
    }
protected:
    lmdb_env _env;
    callback_type _callback;

    int handle_command(Ref<IInteractiveTrans> dev, callback_type callback)
    {
        namedpipe_cmdadapter adapter(dev);
        StringArguments a;
        if(!adapter.recv(a)) return DeviceError::RecvErr;
        string cmd;
        string arg;
        if(!a.Get("Command", cmd) || !a.Get("Arg", arg))
            return DeviceError::RecvFormatErr;
        if(!_async(cmd, arg, callback))
            return DeviceError::ArgRangeErr;
        StringArguments r;
        r.Put("CODE", "0");
        r.Put("RLT", "");
        if(!adapter.send(r)) return DeviceError::SendErr;
        return DeviceError::Success;
    }

    void async_callback(const string& cmd, bool isOK, const string& param, const string& rlt)
    {
        if(!isOK) return ;
        string key = get_key(cmd, param);
        lmdb_event_agent<semaphore_signal, lmdb_string_converter> agent(key.c_str(), _env);
        agent.write(rlt);
    }
public:
    namedpipe_async_svr() {}
    namedpipe_async_svr(const char* dir) { load(dir); }

    int run(callback_type callback,
        uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        NamedPipeDevice dev;
        if(!dev.Open(".", ZABBIX_ASYNC_NAMEDPIPE_NAME, 2000))
            return DeviceError::DevOpenErr;
        if(!dev.WaitConnect(timeoutMs, DEV_OPERATOR_INTERVAL, interrupter))
            return DeviceError::WaitTimeOutErr;
        return handle_command(dev, callback);
    }

    namedpipe_async_svr& cache(lmdb_env& env)
    {
        _env = env;
        _callback = std::bind(&namedpipe_async_svr::async_callback, this,
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4);
        return (*this);
    }
    int run(uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        return run(_callback, timeoutMs, interrupter);
    }
};
//--------------------------------------------------------- 
/**
 * @brief 异步的管道客户端监控项
 * @date 2018-10-12 15:52:21
 */
class namedpipe_async_monitor
{
public:
    typedef typename namedpipe_monitor::value_type value_type;

    static int run(const char* lmdb_dir, const char* cmd, const char* arg, string& rlt, 
        uint timeoutMs = DEV_WAIT_TIMEOUT, 
        uint intervalMs = DEV_OPERATOR_INTERVAL,
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        int err = namedpipe_monitor::run(cmd, arg, rlt, timeoutMs, intervalMs, interrupter, ZABBIX_ASYNC_NAMEDPIPE_NAME);
        if(err != DeviceError::Success) return err;
        // 使用lmdb事件监听实际的结果数据
        string key = namedpipe_async_svr::get_key(cmd, arg);
        lmdb_event_agent<semaphore_signal, lmdb_string_converter> agent(key.c_str(), lmdb_dir);
        bool bOK = agent.wait(rlt, timeoutMs, interrupter);
        agent.remove_read();
        return bOK ? DeviceError::Success : DeviceError::OperatorErr;
    }
    /**
     * @brief 命令行执行监控项
     * @date 2018-10-13 11:36:08
     *
     * argv[1] : lmdb_dir lmdb数据库目录
     * argv[2] : cmd 调用的监控命令
     * argv[*] : arg 监控命令的命令行参数
     */
    static value_type execute(int argc, const char* argv[])
    {
        if(argc < 1) return namedpipe_monitor::put_err(DeviceError::ArgLengthErr);
        int offset = 1;
        const char* lmdb_dir = argv[offset++];
        const char* cmd = argv[offset++];
        string arg = "";
        for(int i = offset; i < argc; ++i)
        {
            arg += argv[i];
            arg += ' ';
        }
        if(argc > offset) arg.pop_back();
        string rlt = "";
        int err = run(lmdb_dir, cmd, arg.c_str(), rlt);
        if(err == DeviceError::Success) return rlt;
        return namedpipe_monitor::put_err(err);
    }
};
//--------------------------------------------------------- 
} // namespace zabbix_extractor 
} // namespace wrapper
} // namespace zhou_yb 
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_NAMEDPIPE_ASYNC_SVR_H_
//========================================================= 